package drds.server.handler;

import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.api_warpper.Lock;
import drds.server.datasource.IConnectionHolder;
import drds.server.net.packet.ErrorPacket;

/**
 * heartbeat check for mysql connections
 */
public class ConnectionHeartBeatHandler extends HandlerAdapter {
	private static final Logger LOGGER = LoggerFactory.getLogger(ConnectionHeartBeatHandler.class);
	protected final Lock lock = new Lock();
	private final ConcurrentHashMap<Long, HeartBeatConnectionHolder> heartBeatConnectionHolderMap = new ConcurrentHashMap<Long, HeartBeatConnectionHolder>();

	public void doHeartBeat(IConnectionHolder connectionHolder, String sql) {
		try {

			HeartBeatConnectionHolder heartBeatConnectionHolder = new HeartBeatConnectionHolder(connectionHolder);
			boolean notExist = (heartBeatConnectionHolderMap.putIfAbsent(heartBeatConnectionHolder.connectionHolder.getId(), heartBeatConnectionHolder) == null);
			if (notExist) {
				connectionHolder.setHandler(this);
				connectionHolder.query(sql);

			}
		} catch (Exception e) {
			executeException(connectionHolder, e);
		}
	}

	/**
	 * remove timeout connections
	 */
	public void abandTimeOuttedConns() {
		if (heartBeatConnectionHolderMap.isEmpty()) {
			return;
		}
		Collection<IConnectionHolder> connectionHolderList = new LinkedList<IConnectionHolder>();
		long curTime = System.currentTimeMillis();
		Iterator<Entry<Long, HeartBeatConnectionHolder>> iterator = heartBeatConnectionHolderMap.entrySet().iterator();
		while (iterator.hasNext()) {
			HeartBeatConnectionHolder heartBeatConnectionHolder = iterator.next().getValue();
			if (heartBeatConnectionHolder.timeOutTimestamp < curTime) {
				connectionHolderList.add(heartBeatConnectionHolder.connectionHolder);
				iterator.remove();
			}
		}

		if (!connectionHolderList.isEmpty()) {
			for (IConnectionHolder connectionHolder : connectionHolderList) {
				try {
					// if(con.isBorrowed())
					connectionHolder.close("heartbeat timeout ");
				} catch (Exception e) {
					LOGGER.error("close err:" + e);
				}
			}
		}

	}

	@Override
	public void errorResponse(IConnectionHolder connectionHolder, byte[] errorBytes) {
		removeFinished(connectionHolder);
		ErrorPacket err = new ErrorPacket();
		err.read(errorBytes);
		LOGGER.error("errorResponse " + err.errorNo + " " + new String(err.errorMessage));
		connectionHolder.release();

	}

	@Override
	public void okResponse(IConnectionHolder connectionHolder, byte[] okBytes) {
		boolean executeResponse = connectionHolder.syncAndExcute();
		if (executeResponse) {
			removeFinished(connectionHolder);
			connectionHolder.release();
		}

	}

	@Override
	public void rowEofResponse(IConnectionHolder connectionHolder, byte[] eofBytes) {
		removeFinished(connectionHolder);
		connectionHolder.release();
	}

	private void executeException(IConnectionHolder connectionHolder, Throwable e) {
		removeFinished(connectionHolder);
		LOGGER.error("executeException   ", e);
		connectionHolder.close("heatbeat exception:" + e);

	}

	private void removeFinished(IConnectionHolder connectionHolder) {
		Long id = ((IConnectionHolder) connectionHolder).getId();
		this.heartBeatConnectionHolderMap.remove(id);
	}

	@Override
	public void connectionClose(IConnectionHolder connectionHolder, String reason) {
		removeFinished(connectionHolder);
		LOGGER.error("connection closed " + connectionHolder + " reason:" + reason);
	}

}

class HeartBeatConnectionHolder {
	public final long timeOutTimestamp;
	public final IConnectionHolder connectionHolder;

	public HeartBeatConnectionHolder(IConnectionHolder connectionHolder) {
		super();
		this.timeOutTimestamp = System.currentTimeMillis() + 20 * 1000L;
		this.connectionHolder = connectionHolder;
	}

}